package com.ktjy.iottemp.rabbitmq.consumer;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.ktjy.temp.pojo.DeviceData;
import com.ktjy.temp.pojo.MqttMessage;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;

@Component
@Slf4j
public class EmqxTempConsumer {

    @Autowired
    private ObjectMapper objectMapper;

    // 使用Rabbit
    // Listener注解监听指定队列
    @RabbitListener(queues = "emqx_temp_queue", containerFactory = "rabbitListenerContainerFactory")
    public void handleMessage(Message message, Channel channel) throws IOException {
        //获取当前消息的唯一标识（投递标签），用于在手动确认模式下告诉 RabbitMQ 具体要确认或拒绝哪条消息。
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        String json = new String(message.getBody());
        try {
            System.out.println("接收到的消息: " + json);

            // 解析外层消息
            MqttMessage mqttMessage = objectMapper.readValue(json, MqttMessage.class);

            // 解析内层负载
            DeviceData payload = objectMapper.readValue(mqttMessage.getPayload(), DeviceData.class);

            log.info("用户名: " + mqttMessage.getUsername());
            log.info("主题: " + mqttMessage.getTopic());
            log.info("payload: " + payload);

            // 模拟业务处理
            Thread.sleep(4000);
            //拿到数据进行业务处理，正好数据库连接异常抛出异常
            //throw new RuntimeException("数据库连接异常");
            if (payload.getTemperature() > 30) {
                System.out.println("温度过高，发送邮件提醒");
                System.out.println("发送邮件提醒成功");
                throw new RuntimeException("温度过高");
            }
            // 业务处理成功，手动确认消息
            // 第二个参数：是否批量确认
            //channel.basicAck(deliveryTag, false);
            System.out.println("消息处理成功，已手动确认，deliveryTag: " + deliveryTag);

        } catch (Exception e) {
            System.err.println("消息处理失败: " + e.getMessage());
            // 处理失败，根据需求决定是否重新入队
            // 第三个参数：是否重新入队
            //channel.basicNack(deliveryTag, false, true);
            // 如果不希望重新入队，可以使用basicReject
            //channel.basicReject(deliveryTag, false);
            System.out.println("消息处理失败，已拒绝，deliveryTag: " + deliveryTag);
        }
    }
}